package com.cmb.lk50.common;

import com.cmb.lk50.pojo.AbstractLog;
import com.cmb.lk50.sink.LogSink;
import com.cmb.lk50.sink.kafka.KafkaKeyGenerator;
import com.cmb.lk50.sink.kafka.KafkaSink;

import java.util.List;

/**
 * Author       : 蒋俊钊 80256973
 * Date         : 2020/8/25
 * Copyright    (C) ChinaMerchantsBank
 */
class Sinkable {

    void doSinks(List<LogSink> logSinks, AbstractLog log, KafkaKeyGenerator<AbstractLog> keyGenerator) {
        if (logSinks != null) {
            logSinks.forEach(sink -> {
                if (sink instanceof KafkaSink) {
                    KafkaSink kafkaSink = (KafkaSink) sink;
                    kafkaSink.writeWithKey(log, keyGenerator);
                } else {
                    sink.write(log);
                }
            });
        }
    }
}
